/*
 * AISMessages
 * - a java-based library for decoding of AIS messages from digital VHF radio traffic related
 * to maritime navigation and safety in compliance with ITU 1371.
 *
 * (C) Copyright 2011--3 by S-Consult ApS, DK31327490, http://tbsalling.dk, Denmark.
 *
 * Released under the Creative Commons Attribution-NonCommercial-ShareAlike 3.0 Unported License.
 * For details of this license see the nearby LICENCE-full file, visit http://creativecommons.org/licenses/by-nc-sa/3.0/
 * or send a letter to Creative Commons, 171 Second Street, Suite 300, San Francisco, California, 94105, USA.
 *
 * NOT FOR COMMERCIAL USE!
 * Contact Thomas Borg Salling <tbsalling@tbsalling.dk> to obtain a commercially licensed version of this software.
 *
 */

package com.xx.dk.tbsalling.aismessages.nmea;

import com.xx.dk.tbsalling.aismessages.ais.messages.AISMessage;
import com.xx.dk.tbsalling.aismessages.nmea.messages.NMEAMessage;
import com.xx.dk.tbsalling.aismessages.utils.LimitedSizeMap;
import com.xx.dk.tbsalling.aismessages.ais.messages.Metadata;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.function.Consumer;

/**
 * 此类接收包含铠装和编码的 AIS 字符串的 NMEA 消息。
 * 一条 AIS 消息可以跨越多条 NMEA 消息。每当完整的 AIS 消息
 * 时，将构造一个 AISMessage 并将其传递给所有已注册的
 * 编码的 AIS 消息的接收者。
 *
 * @author tbsalling
 */
@Slf4j
public class NMEAMessageHandler implements Consumer<NMEAMessage> {

    //private static final System.Logger LOG = System.getLogger(NMEAMessageHandler.class.getName());

    private final String source;
    private final ArrayList<NMEAMessage> messageFragments = new ArrayList<>();
    private static final LimitedSizeMap<String, ArrayList<NMEAMessage>> MESSAGEFRAGMENTSMAP = new LimitedSizeMap<>(10);
    private final List<Consumer<? super AISMessage>> aisMessageReceivers = new LinkedList<>();
    private final boolean bool = true;

    public NMEAMessageHandler(String source, Consumer<? super AISMessage>... aisMessageReceivers) {
        this.source = source;
        for (Consumer<? super AISMessage> aisMessageReceiver : aisMessageReceivers) {
            addAisMessageReceiver(aisMessageReceiver);
        }
    }

    /**
     * Receive a single NMEA amoured AIS string
     *
     * @param nmeaMessage the NMEAMessage to handle.
     */
    @Override
    public void accept(NMEAMessage nmeaMessage) {
        if (!nmeaMessage.isValid()) {
            log.warn("NMEA message is invalid: " + nmeaMessage.toString());
            return;
        }

        int numberOfFragments = nmeaMessage.getNumberOfFragments();
        if (numberOfFragments <= 0) {
            log.warn("NMEA message is invalid: " + nmeaMessage.toString());
            messageFragments.clear();
        } else if (numberOfFragments == 1) {
            log.debug("Handling unfragmented NMEA message");
            AISMessage aisMessage = AISMessage.create(new Metadata(source), nmeaMessage.getTagBlock(), nmeaMessage);
            sendToAisMessageReceivers(aisMessage);
            messageFragments.clear();
        } else {
            int fragmentNumber = nmeaMessage.getFragmentNumber();
            log.debug("Handling fragmented NMEA message with fragment number " + fragmentNumber);
            if (fragmentNumber < 0) {
                log.warn("Fragment number cannot be negative: " + fragmentNumber + ": " + nmeaMessage.getRawMessage());
                messageFragments.clear();
            } else if (fragmentNumber > numberOfFragments) {
                log.debug("Fragment number " + fragmentNumber + " higher than expected " + numberOfFragments + ": " + nmeaMessage.getRawMessage());
                messageFragments.clear();
            } else {
                if (bool) { //Multiple AISMessage needs to be parsed even if the sequence is inconsistent
                    String[] msg = nmeaMessage.getRawMessage().split(",");
                    String key = msg[1] + msg[3] + msg[4];
                    ArrayList<NMEAMessage> nmeaMessages = MESSAGEFRAGMENTSMAP.get(key);
                    if (nmeaMessages == null) {
                        if (fragmentNumber == 1) {
                            nmeaMessages = new ArrayList<>();
                            nmeaMessages.add(nmeaMessage);
                            MESSAGEFRAGMENTSMAP.put(key, nmeaMessages);
                        }
                    } else {
                        nmeaMessages.add(nmeaMessage);
                        if ( Integer.valueOf(msg[1]) == nmeaMessages.size()) {
                            AISMessage aisMessage = AISMessage.create(new Metadata(source), nmeaMessage.getTagBlock()
                                    , nmeaMessages.toArray(new NMEAMessage[nmeaMessages.size()]));
                            sendToAisMessageReceivers(aisMessage);
                            MESSAGEFRAGMENTSMAP.remove(key);
                        } else
                            log.debug("Fragmented message not yet complete; missing " + (Integer.valueOf(msg[1]) - nmeaMessages.size()) + " fragment(s).");
                    }
                } else {
                    int expectedFragmentNumber = messageFragments.size() + 1;
                    log.debug("Expected fragment number is: " + expectedFragmentNumber + ": " + nmeaMessage.getRawMessage());

                    if (expectedFragmentNumber != fragmentNumber) {
                        log.debug("Expected fragment number " + expectedFragmentNumber + "; not " + fragmentNumber +
                                ": " + nmeaMessage.getRawMessage());
                        messageFragments.clear();
                    } else {
                        messageFragments.add(nmeaMessage);
                        log.debug("nmeaMessage.getNumberOfFragments(): " + nmeaMessage.getNumberOfFragments());
                        log.debug("messageFragments.size(): " + messageFragments.size());
                        if (nmeaMessage.getNumberOfFragments() == messageFragments.size()) {
                            AISMessage aisMessage = AISMessage.create(new Metadata(source), nmeaMessage.getTagBlock()
                                    , messageFragments.toArray(new NMEAMessage[messageFragments.size()]));
                            sendToAisMessageReceivers(aisMessage);
                            messageFragments.clear();
                        } else
                            log.debug("Fragmented message not yet complete; missing " + (nmeaMessage.getNumberOfFragments() - messageFragments.size()) + " fragment(s).");
                    }
                }
            }
        }
    }

    /**
     * Send encoded AIS message to all interested receivers.
     */
    private void sendToAisMessageReceivers(final AISMessage aisMessage) {
        aisMessageReceivers.forEach(r -> r.accept(aisMessage));
    }

    /**
     * Add a consumer of encoded AIS messages.
     *
     * @param aisMessageReceiver The consumer to add.
     */
    @SuppressWarnings("unused")
    public void addAisMessageReceiver(Consumer<? super AISMessage> aisMessageReceiver) {
        aisMessageReceivers.add(aisMessageReceiver);
    }

    /**
     * Empty buffer of unhandled messages and return those not handled.
     *
     * @return List of unhandled NMEAMessages.
     */
    @SuppressWarnings("unchecked")
    public ArrayList<NMEAMessage> flush() {
        ArrayList<NMEAMessage> unhandled = (ArrayList<NMEAMessage>) messageFragments.clone();
        messageFragments.clear();
        return unhandled;
    }

}
